package com.xiaoze.testapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/common/
 *
 * @author xiaoze
 * @date 2021/12/19
 */
public class CommonApi {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 从文件中读取数据
        String path = System.getProperty("user.dir") + "\\src\\main\\resources\\sensor.txt";

        TableDescriptor sourceDescriptor = TableDescriptor.forConnector("filesystem")
                .option("path", path)
                .format(FormatDescriptor.forFormat("csv")
                        .option("field-delimiter", ",")
                        .build())
                .schema(Schema.newBuilder()
                        .column("id", DataTypes.STRING())
                        .column("timestamp", DataTypes.DOUBLE())
                        .column("temp", DataTypes.DOUBLE())
                        .build())
                .build();

        tableEnv.createTemporaryTable("inputTable", sourceDescriptor);

        // 得到表
        Table inputTable = tableEnv.from("inputTable");
        tableEnv.toDataStream(inputTable).print();

        env.execute();

    }

}
